package org.ala.tiktools.rpc.kcp.protocol.receiver;

import java.util.List;

import org.ala.tiktools.rpc.kcp.protocol.IKcp;
import org.ala.tiktools.rpc.kcp.protocol.IKcpReceiverQueue;
import org.ala.tiktools.rpc.kcp.protocol.Kcp;
import org.ala.tiktools.rpc.kcp.protocol.receiver.queue.KcpReceiveArrayQueue;
import org.ala.tiktools.rpc.kcp.protocol.segment.KcpSegment;
import org.ala.tiktools.tools.Assert;
import org.ala.tiktools.tools.queue.IQueueIterator;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;

/**
 * 接受者队列
 *
 * @author ala
 * @date 2025-04-11 22:28
 */
public class KcpReceiver2 {
    /**
     *  绑定的kcp
     */
    protected Kcp kcp;


    /**
     *  下一个接收的sn
     *  <p> 从1开始，0流出来判断绕回
     *  <p> 保证消息的有序接收
     *  <p> rcvBuf按照sn号底层顺序插入
     *  <p> rcvQue按照sn号递增写给应用层buf
     */
    protected long rcvNxt = 1L;
    /**
     *  rcvWnd
     */
    protected int rcvWnd;
    /**
     *  接收缓冲区
     */
    protected IKcpReceiverQueue rcvBuf;
    /**
     * 	发生绕回前的接收缓冲区
     * 	<p>	在缓冲区为空时释放掉
     */
    protected IKcpReceiverQueue snWarpRoundRcvBuf;
    
    /**
     *  接收队列
     */
    protected IKcpReceiverQueue rcvQue;
    /**
     *  ByteBuf内存分配器
     */
    protected ByteBufAllocator byteBufAllocator = ByteBufAllocator.DEFAULT;


    public KcpReceiver2(Kcp kcp, int rcvWnd) {
        this.kcp = kcp;
        this.rcvWnd = rcvWnd;
//        rcvBuf = new KcpReceiveLinkedQueue();
//        rcvQue = new KcpReceiveLinkedQueue();
      rcvBuf = new KcpReceiveArrayQueue(rcvWnd);
      rcvQue = new KcpReceiveArrayQueue(rcvWnd);		//	貌似是他的问题
    }


    /**
     *  读取数据给上层应用 rcvQue -> buf
     *  @return 高32位：返回的状态码 | 低32位：读取的字节数<br/>
     *          高32位：2:需要询问对方窗口大小 | 0:不需要<br/>
     *          第32位：-1:queue为空 | -2:queue中还么有收到一个完整消息 | >0 一个完整消息的字节数<br/>
     */
    public long recv(ByteBuf buf) {
        //  检测当前rcvQue中是否包含一个完整消息
        int peekSize = rcvQue.peekSize();
        if (peekSize < 0) {
            return peekSize;
        }

        //  是否需要询问对方窗口大小
        boolean recover = rcvQue.isWarningLine(rcvWnd);

        //  rcvQue -> buf
        int len = rcvQue.writeToByteBufAndRemove(buf);
        Assert.assertTrue(len == peekSize, String.format("rcvQue在读取时发生了改变，读取前后长度不一致 before:%d after:%d", peekSize, len));

        //  判断是否需要询问对方窗口
        int probe = 0;
        if (!rcvQue.isWarningLine(rcvWnd) && recover) {
            probe = IKcp.IKCP_ASK_TELL;
        }

        //  rcvBuf -> rcvQue
        bufMoveToQue();

        return (long)probe << 32 | len;
    }


    /**
     *  rcvQue 中第一个完整消息的字节数
     *  @return -1:queue为空<br/>
     *          -2:queue中还么有收到一个完整消息<br/>
     *          >0 一个完整消息的字节数
     */
    public int peekSize() { 
    	return rcvQue.peekSize(); 
    }
    /**
     *  是否可以接收数据
     */
    public boolean canRecv() {
        if (rcvQue.isEmpty()) {
            return false;
        }
        KcpSegment seg = rcvQue.first();
        if (seg.getFrg() == 0) {
            return true;
        }
        return rcvQue.size() >= seg.getFrg() + 1;
    }
    /**
     *  读取数据给上层应用 rcvQue[i] -> buf[i]
     *  @return 高32位：返回的状态码 | 低32位：读取的字节数<br/>
     *          高32位：2:需要询问对方窗口大小 | 0:不需要<br/>
     *          第32位：-1:queue为空 | -2:queue中还么有收到一个完整消息 | >0 一个完整消息的字节数<br/>
     */
    public long recv(List<ByteBuf> bufs) {
        //  检测当前rcvQue中是否包含一个完整消息
        int peekSize = rcvQue.peekSize();
        if (peekSize < 0) {
            return peekSize;
        }

        //  是否需要询问对方窗口大小
        boolean recover = rcvQue.isWarningLine(rcvWnd);

        //  rcvQue -> buf
        int len = rcvQue.writeToByteBufsAndRemove(bufs);
        Assert.assertTrue(len == peekSize, String.format("rcvQue在读取时发生了改变，读取前后长度不一致 before:%d after:%d", peekSize, len));

        //  判断是否需要询问对方窗口
        long probe = 0;
        if (!rcvQue.isWarningLine(rcvWnd) && recover) {
            probe = IKcp.IKCP_ASK_TELL;
        }

        //  rcvBuf -> rcvQue
        bufMoveToQue();

        return probe << 32 | len;
    }
    /**
     *  将接收到的数据读给 rcvBuf
     *  @return 高32位：返回的状态码 | 低32位：读取的字节数<br/>
     *          高32位：<br/>
     *          低32位：-1:buf中的数据本来就不够读一个kcp头（1个都没读的还）<br/>
     *                 -2:非法数据（可读区域不够1个数据段，或者数据段长度 < 0）<br/>
     *                 -3:非法消息（cmd不可识别）<br/>
     *                 -4:conv不是发给本conv的（收到了别人的数据）<br/>
     *                 >:接收成功，并且返回本次收到的最大ackSn
     */
    public long input(ByteBuf buf) {
        //  记录本次input收到的最大ack的sn
        long maxAckSn = 0;
        //  返回的高32位
        int probe = 0;
        //  读之前的sndUna（如果发生了改变，需要更新拥塞窗口）
//        long oldSndUna = kcp.getSender().getSndUna();

        int rtn = 0;
        //  连续读，直到读不到1个消息头为止
        while (buf.readableBytes() >= IKcp.IKCP_OVERHEAD) {
            //  读kcp头
            int conv = buf.readIntLE();
            byte cmd = buf.readByte();
            short frg = buf.readUnsignedByte();
            int wnd = buf.readUnsignedShortLE();
            int ts = buf.readIntLE();
            long sn = buf.readUnsignedIntLE();
            long una = buf.readUnsignedIntLE();
            int len = buf.readIntLE();

            //  如果剩下的数据不够 len，直接-2
            if (len < 0 || buf.readableBytes() < len) { 
            	rtn = -2; 
            	break; 
            }
            //  如果cmd非法，直接-3
            if (cmd != IKcp.IKCP_CMD_PUSH && cmd != IKcp.IKCP_CMD_ACK && cmd != IKcp.IKCP_CMD_WASK && cmd != IKcp.IKCP_CMD_WINS) { rtn = -3; break; }
            //  是否自动conv
            if (conv == 0 && kcp.isAutoSetConv()) { kcp.setConv(conv); }
            //  检测conv是否是本conv
            if (conv != kcp.getConv()) { rtn = -4; break; }

            //  更新远程窗口
            kcp.setRmtWnd(wnd);

            //  更新sndUna
            kcp.getSender().parseUna(una);
            kcp.getSender().shrinkBuf();

            //  根据cmd处理消息
            int current = kcp.getCurrent();
            switch (cmd) {
                case IKcp.IKCP_CMD_ACK: {
                    //  重新计算rto（rtt = 当前时间戳 - 当时发消息的时间戳）
                    kcp.getRto().updateAck(current - ts);
                    //  确认sn收到了
                    kcp.getSender().parseAck(sn);
                    kcp.getSender().shrinkBuf();
                    //  记录本次收到的最大sn
                    maxAckSn = Math.max(maxAckSn, sn);
                    //  对于ack消息，数据段不需要读取，直接跳过
                    if (len > 0) { buf.skipBytes(len); }
                    break;
                }
                case IKcp.IKCP_CMD_PUSH: {
                    if (sn < rcvNxt + rcvWnd) {
                        //  追加待回复
                        kcp.getAckList().add((int)sn, ts);
                        if (sn >= rcvNxt) {
                            KcpSegment seg = null;
                            if (len > 0) {
                            	//	0-copy
//                            	seg = KcpSegment.alloc(byteBufAllocator, len);
//                            	seg.getData().readBytes(buf, buf.readerIndex(), len);
                                seg = KcpSegment.alloc(buf.readRetainedSlice(len));
                            } else {
                                seg = KcpSegment.alloc(byteBufAllocator, 0);
                            }
                            seg.setConv(conv);
                            seg.setCmd(cmd);
                            seg.setFrg(frg);
                            seg.setWnd(wnd);
                            seg.setTs(ts);
                            seg.setSn(sn & 0xFFFFFFFFL);    //  转32位无符号
                            seg.setUna(una);
                            seg.setLen(len);
                            //  解析收到的信息，并保存进rcvBuf
                            parseData(seg);
                        }
                    }
                    break;
                }
                case IKcp.IKCP_CMD_WASK: {
                    probe |= IKcp.IKCP_ASK_TELL;
                    //  对于询问窗口消息，数据段不需要读取，直接跳过
                    if (len > 0) { buf.skipBytes(len); }
                    break;
                }
                case IKcp.IKCP_CMD_WINS: {
                    //  无事发生，上面已经接更新过远程窗口了
                    break;
                }
            }
        }
        //  如果执行失败，高32位是状态码，直接返回
        if (rtn < 0) {
            return (long)probe << 32 | rtn;
        }
        //  否则高32位是本次收到的最大ackSn
        else {
            return (long)probe << 32 | maxAckSn;
        }
    }


    /**
     *  rcvBuf -> rcvQue，但不能超过rmtWnd
     *  @return 导了多少数据
     */
    private int bufMoveToQue() {
        int len = 0;
        
        //	如果绕回前队列不为空，优先从绕回前队列取数
        if (snWarpRoundRcvBuf != null) {
        	//	不再判断sn号是否连续，老队列既然能入队，一定是连续的
        	IQueueIterator<KcpSegment> iterator = snWarpRoundRcvBuf.firstIterator();
        	while (iterator.hasNext()) {
        		rcvQue.addFirst(iterator.removeAndNext());
        		len++;
        	}
        	snWarpRoundRcvBuf = null;
        }
        //	否则从rcvbuf中取数
        IQueueIterator<KcpSegment> iterator = rcvBuf.firstIterator();
        KcpSegment seg = null;
        while (iterator.hasNext() && (seg = iterator.current()).getSn() == rcvNxt++ && !rcvQue.isWarningLine(rcvWnd)) {
        	rcvQue.addLast(seg);
        	len++;
        	iterator.removeAndNext();
        }
        return len;
    }

    /**
     *  将受到的数据按sn号升序放入sndBuf列表
     *  <p> sn号大概率收到的是排在末尾的
     */
    private void parseData(KcpSegment seg) {
        long sn = seg.getSn();
        //	如果sn号发生了绕回，新建rcvBuf用于接收消息，原rcvBuf在为空时释放掉。原rcvBuf中的sn号相对绕回后太大了，无法维护sn顺序
        if (sn == 0) {
        	rcvNxt = sn;
        	snWarpRoundRcvBuf = rcvBuf;
//        	rcvBuf = new KcpReceiveLinkedQueue();
        	rcvBuf = new KcpReceiveArrayQueue(rcvWnd);
        }
        
        if (sn >= rcvNxt + rcvWnd || sn < rcvNxt) {
            return;
        }
        rcvBuf.parseData(seg);
        //  尝试移动数据到rcvQue
        bufMoveToQue();
    }

    /**
     *  接收窗口中的剩余容量
     */
    public int wndUnused() {
        return Math.max(0, rcvWnd - rcvQue.size());
    }
    /**
     *  下一个要接收的sn
     */
    public long getRcvNxt() {
        return this.rcvNxt;
    }


    /**
     *  释放
     */
    public void release() {
    	rcvBuf.clear();
    	rcvQue.clear();
    }

    public int getRcvWnd() {
        return rcvWnd;
    }

    public void setRcvWnd(int rcvWnd) {
        this.rcvWnd = rcvWnd;
    }


	public IKcpReceiverQueue getRcvBuf() {
		return rcvBuf;
	}


	public void setRcvBuf(IKcpReceiverQueue rcvBuf) {
		this.rcvBuf = rcvBuf;
	}


	public IKcpReceiverQueue getRcvQue() {
		return rcvQue;
	}


	public void setRcvQue(IKcpReceiverQueue rcvQue) {
		this.rcvQue = rcvQue;
	}
}
